package my.yuan.ms.rediskafkams.consumer;

import lombok.extern.slf4j.Slf4j;
import my.yuan.ms.rediskafkams.controller.vo.MsVO;
import my.yuan.ms.rediskafkams.service.MsService;
import my.yuan.ms.rediskafkams.service.StockService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

import static my.yuan.ms.rediskafkams.producer.KafkaProducer.GROUP_ID;
import static my.yuan.ms.rediskafkams.producer.KafkaProducer.TOPIC_DEMO;

/**
 * @author zzMh
 * @Description TODO
 */
@Component
@Slf4j
public class KafkaConsumer {
    @Autowired
    private MsService msService;
    @Autowired
    private StockService stockService;

    @KafkaListener(topics = {TOPIC_DEMO}, groupId = GROUP_ID, containerFactory = "kafkaListenerContainerFactory",concurrency = "1")
    public void consumeMessage(//ConsumerRecord<String, MsVO> record,
                               List<ConsumerRecord<String,MsVO>> list,
                               //@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) List<Integer> keys,
                               //@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                               //@Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
                               //@Header(KafkaHeaders.OFFSET) List<Long> offsets,
                               Acknowledgment ack) {
       try {
           for (int i = 0; i < list.size(); i++) {
               ConsumerRecord<String, MsVO> record = list.get(i);
               try {
                   MsVO msVO = record.value();
                   msService.handleMsAndOrder(msVO);

               } catch (Exception e) {
                   //mysql事务执行失败，回滚库存
                   log.error("mysql事务执行失败，回滚redis库存");
                   stockService.redisPlusStock(record.value().getCid());
               }
           }
       }finally {
            ack.acknowledge();
        }


    }
}
